package com.laycoding.rocketmq.producer.service.impl;

import com.laycoding.rocketmq.producer.entity.RocketMqMessage;
import com.laycoding.rocketmq.producer.service.RocketMqService;
import lombok.extern.log4j.Log4j2;
import org.apache.rocketmq.client.producer.SendCallback;
import org.apache.rocketmq.client.producer.SendResult;
import org.apache.rocketmq.spring.core.RocketMQTemplate;
import org.springframework.messaging.support.MessageBuilder;
import org.springframework.stereotype.Service;

import javax.annotation.Resource;


/**
 * created with IntelliJ IDEA.
 *
 * @author: guidada
 * @date: 2021/04/09/11:00
 * @description:
 */
@Service("rocketMqService")
@Log4j2
public class RocketMqServiceImpl implements RocketMqService {
    @Resource
    RocketMQTemplate rocketMQTemplate;

    @Override
    public void send(RocketMqMessage mqMsg) {
        log.info("send发送消息到mqMsg={}", mqMsg);
        rocketMQTemplate.send(mqMsg.getTopic() + ":" + mqMsg.getTags(),
                MessageBuilder.withPayload(mqMsg.getContent()).build());
    }

    @Override
    public void asyncSend(RocketMqMessage mqMsg) {
        log.info("asyncSend发送消息到mqMsg={}", mqMsg);
        rocketMQTemplate.asyncSend(mqMsg.getTopic() + ":" + mqMsg.getTags(), mqMsg.getContent(),
                new SendCallback() {
                    @Override
                    public void onSuccess(SendResult sendResult) {
                        // 成功不做日志记录或处理
                    }

                    @Override
                    public void onException(Throwable throwable) {
                        log.info("mqMsg={}消息发送失败", mqMsg);
                    }
                });
    }

    @Override
    public void syncSendOrderly(RocketMqMessage mqMsg) {
        log.info("syncSendOrderly发送消息到mqMsg={}", mqMsg);
        rocketMQTemplate.sendOneWay(mqMsg.getTopic() + ":" + mqMsg.getTags(), mqMsg.getContent());
    }
}
